package com.teamsun.mqtt.service;

import io.netty.handler.codec.mqtt.MqttQoS;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.atomic.AtomicInteger;

import org.apache.ibatis.session.SqlSession;

import com.teamsun.mqtt.db.DBSessionFactory;
import com.teamsun.mqtt.entity.MsgRep;
import com.teamsun.mqtt.entity.SendMsg;

public class GetUnReadData implements Callable<Void>{

	
	BlockingQueue<MsgRep> msgRepsque;
	/**
	 * 准备更改状态的 已发送消息
	 */
	BlockingQueue<SendMsg> readySendedQue;
	
	static  AtomicInteger atomicInteger=new AtomicInteger(1);
	
	static  AtomicInteger successInteger=new AtomicInteger(1);
	
	static final   String  selectUnreadMsgByMsgIds="com.teamsun.mqtt.entity.MsgRepMapper.selectUnreadMsgByMsgIds";
	
	static final  String selectUnReadUserMsgIds="com.teamsun.mqtt.entity.MsgRepMapper.selectUnReadUserMsgIds"; 
	
	
	
	String clientid;
	public   GetUnReadData(String clientid){
		this.clientid=clientid;
		this.readySendedQue=MessageDataService.readySendedQue;
		this.msgRepsque=MessageDataService.msgRepsque;
	}
	
	/**
	 *  一共四种情况。
	 *  TAB_send<--->TAB_msg
	 *  TAB_send<--->MEM_msg
	 *  MEM_sen<--->TAB_msg
	 *  MEM_sen<--->MEM_msg
	 */
	@Override
	public Void call() throws Exception {
	
		
		SqlSession session=DBSessionFactory.getSqlSession();
		
		System.out.println(""+ atomicInteger.incrementAndGet()+"个线程在查询数据库");
		List<MsgRep> msgReps=
				queryBySendTabGetMsg(session);
		
		msgReps.addAll(queryMsgTabGetMsg(session));
		session.close();
		for(MsgRep msgRep:msgReps){
			
			if(msgRep.getContent()!=null)
				
			MessageDataService.sendMsg2Client(msgRep, clientid,MqttQoS.EXACTLY_ONCE);
		}
		 System.out.println(successInteger.incrementAndGet()+"个线程成功");
		return null;	
		
	}
	
	
	
	/**
	 * 从readySendedQue 里面获取待收消息id  分别在消息队列和数据库取出来消息
	 * @param session
	 * @return
	 */
	public  List<MsgRep> queryMsgTabGetMsg(SqlSession session){
		
		List<MsgRep> result=new ArrayList<MsgRep>();
		Iterator<SendMsg> iterator=readySendedQue.iterator();
		List<Integer> msgids=new ArrayList<Integer>();
		while(iterator.hasNext()){
			SendMsg sendMsg=iterator.next();
			if(sendMsg.getClientid().equals(clientid)){
				msgids.add(sendMsg.getSendmsgid());
				readySendedQue.remove(sendMsg);
			}
		}

		if(msgids.isEmpty())
			return result;
		//MEM_send<--->TAB_msg
		Map<String, Object> map=new HashMap<String, Object>();
		map.put("list", msgids);
		result=session.selectList(selectUnreadMsgByMsgIds,map);
		
		Iterator<MsgRep> iteratormsg=
				msgRepsque.iterator();
		 //MEM_send<--->MEM_msg
		while(iteratormsg.hasNext()){
			
			MsgRep msgRep=iteratormsg.next();
			if(msgids.contains(msgRep.getId())){
				result.add(msgRep);
			}
		}
		
		
		return result;
	}
	
	/**
	 * 从已发消息 里面获取待收消息id  分别在消息队列和数据库取出来消息
	 * @param session
	 * @return
	 */
	public  List<MsgRep> queryBySendTabGetMsg(SqlSession session){
		
		
		List<MsgRep> result=new ArrayList<MsgRep>();
		SendMsg msg=new SendMsg();
		msg.setClientid(clientid);
		
		List<Integer> msgids= session.selectList(selectUnReadUserMsgIds,msg);
		
		
		if(msgids.isEmpty())
			return result;
		Map<String, Object> map=new HashMap<String, Object>();
		map.put("list", msgids);
		// TAB_send<--->TAB_msg
		
		result=session.selectList(selectUnreadMsgByMsgIds,map);
		
		Iterator<MsgRep> iterator=
				msgRepsque.iterator();
		
		
		 //TAB_send<--->MEM_msg
		while(iterator.hasNext()){
			
			MsgRep msgRep=iterator.next();
			if(msgids.contains(msgRep.getId())){
				result.add(msgRep);
			}
		}
		
		return result;
	}

	
	
}
